package com.itlong.cloud.commons.thread.queue.process.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.itlong.cloud.POJO.DTO.common.IotPubMessageDTO;
import com.itlong.cloud.POJO.PO.SendPushMessageTaskPO;
import com.itlong.cloud.commons.thread.queue.process.queue.SendIotPushMessageQueue;
import com.itlong.cloud.commons.thread.queue.process.receiver.SendPushMessageReceiver;
import com.itlong.cloud.constants.RabbitMQConstant;
import com.itlong.cloud.enumerate.DeviceIotEnum;
import com.itlong.cloud.rabbitmq.config.MessageSender;
import com.itlong.cloud.utils.node.JsonUtil;
import com.itlong.cloud.utils.random.LogicIdUtil;
import org.apache.commons.lang.StringEscapeUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * <desc>
 *      发送轮询指令至iot中心自定义队列消费者
 * </desc>
 *
 * @createDate 2019/08/01
 */
@Service
public class SendPushMessageConsumer {
    private static final Logger LOG = LoggerFactory.getLogger(SendPushMessageConsumer.class);
    @Autowired
    SendIotPushMessageQueue sendIotPushMessageQueue;
    @Autowired
    MessageSender messageSender;
    private ExecutorService cachedThreadPool ; //线程池

    /**
     * <desc>
     *      线程开始轮询
     * </desc>
     *
     * @param workerId
     * @author Jiaqi.X
     * @createDate 2018/04/10
     */
    public void worker(int workerId) {
        SendPushMessageReceiver receiver = new SendPushMessageReceiver(workerId, sendIotPushMessageQueue.getSendIotPushQueue());
        while (true) {
            try {
                //读取队列中的消息
                List<SendPushMessageTaskPO> sends = receiver.receiveMessage();
                sends = sends.stream().collect(Collectors.collectingAndThen(
                        Collectors.toCollection(() -> new TreeSet<>(Comparator.comparing(o -> o.getDeviceUnique()))), ArrayList::new));
                List<Map<String, Object>> messages = new ArrayList<>();
                for (SendPushMessageTaskPO po: sends) {
                    Map<String, Object> message = new HashMap<>();
                    message.put("productKey", po.getProductKey());
                    message.put("topic", po.getTopic());
                    message.put("serviceId","SettingService");
                    message.put("commandName","dataUpdate");
                    message.put("iotFlag",po.getIotFlag());
                    Map<String,Object> huaweiIotMessage = new HashMap<>();
                    huaweiIotMessage.put("sourceForm",po.getSourceForm());
                    huaweiIotMessage.put("cmd",po.getCmd());
                    huaweiIotMessage.put("snr", LogicIdUtil.bussinessId());
                    huaweiIotMessage.put("timestamp",(new Date()).getTime() + "");
                    boolean mark = false;
                    if(StringUtils.isNotBlank(po.getCmd()) && po.getCmd().equals("webcam")){
                        //此处处理实时抓拍和撤消告警相关处理
                        Map<String, Object> customDataByOne = new HashMap<>();
                        Map<String, Object> customDataByTwo = new HashMap<>();
                        //处理实时抓拍
                        if(po.getCommandType().contains("snapshot")){
                            customDataByTwo.put("eventSeq",po.getCommandType().substring(8,po.getCommandType().length()));
                            customDataByOne.put("data",customDataByTwo);
                            customDataByOne.put("cmd","snapshot");
                            huaweiIotMessage.put("customData", JsonUtil.toJSON(customDataByOne));
                            mark = true;
                        }else if(po.getCommandType().contains("cancelAlarm")){
                            //处理撤消告警
                            customDataByOne.put("data","");
                            customDataByOne.put("cmd","cancelAlarm");
                            huaweiIotMessage.put("customData",JsonUtil.toJSON(customDataByOne));
                            mark = true;
                        }
                        huaweiIotMessage.put("commandType", "");
                    }else{
                        huaweiIotMessage.put("commandType", po.getCommandType().toString());
                    }
                    //mark标识主要处理告警指令下发至IOT会有转义 \
//                    if(mark){
//                        message.put("huaweiIotMessage", StringEscapeUtils.unescapeJava(JsonUtil.toJSON(huaweiIotMessage)));
//                    }else{
                        message.put("huaweiIotMessage",JsonUtil.toJSON(huaweiIotMessage));
//                    }
                    IotPubMessageDTO dto = new IotPubMessageDTO();
                    dto.setSourceForm(po.getSourceForm());
                    dto.setCmd(po.getCmd());
                    dto.setSnr(1);
                    dto.setTimestamp((new Date()).getTime() + "");
                    Map<String, Object> data = new HashMap<>();
                    data.put("commandType", po.getCommandType().toString());
                    dto.setData(data);
                    message.put("message", JsonUtil.toJSON(dto));
                    messages.add(message);
                    LOG.info("发送iot获取轮询指令设备:"+po.getDeviceUnique());
                }
                Map<String,Object> params = new HashMap<>();
                params.put("deviceOperateIotType", DeviceIotEnum.PUB_MESSAGE.getCommandType());
                params.put("devices",messages);
                JSONObject obj = new JSONObject();
                obj.put("data",params);
                messageSender.send(RabbitMQConstant.RABBITMQ_IOT_PUB_QUEUE, obj.toJSONString(),
                        RabbitMQConstant.RABBITMQ_IOT_PUB_EXCHANGE, RabbitMQConstant.RABBITMQ_IOT_PUB_KEY);
            } catch (Exception e) {
                LOG.error("【发送呼梯消息数据异常】",e);
            }
        }
    }

    /**
     * <desc>
     *      初始化
     * </desc>
     *
     * @author Jiaqi.X
     * @createDate 2018/04/10
     */
    @PostConstruct
    public void polling(){
        //初始化线程池
        cachedThreadPool = Executors.newFixedThreadPool(1);
        //开启10个线程消费消息
        for(int i=1;i<=10;i++){
            int id=i;
            cachedThreadPool.execute(() -> {
                worker(id);
            });
        }
    }
}
